//1.创建一个包含100个元素的RDD，将其中大于50的元素乘以2，并筛选出其中大于100的元素。
val data1=sc.makeRDD(1 to 100)
val a=data1.filter(_> 50).map(x=>x*2).filter(_>100)
println(a.collect())

//2.创建一个包含10个元素的RDD，将其中所有元素按空格切分成单词，并将所有单词变成小写形式（转换小写toLowerCase）。
val data2=sc.parallelize(Seq("Spark is good","apple is dilicious","Hadoop","python","Scala","Java","Linux","JavaWeb","HDFS","Yarn"))
val words=data2.flatMap(x=>x.toLowerCase.split(" "))
println(words.collect().mkString(","))


//3.创建一个包含5个元素的RDD，每个元素都是一个二元组，第一个元素是字符串类型的名字，第二个元素是Int类型的分数。
val data3=Array(("Alice",90),("Bob",80),("Charlie",70),("Alice",85),("Bob", 95))
val rdd=sc.parallelize(data3)
val grade=rdd.reduceByKey((a,b)=>a+b)
println(grade.foreach(x=>println(x)))

//4.创建一个包含一些字符串的RDD，例如：
//val rdd = sc.parallelize(Seq("apple", "banana", "orange", "pear", "watermelon", "grape", "pineapple"))
//现在你需要按照字符串长度将其分为两个分区，其中一个分区包含长度小于等于5的字符串，另一个分区包含长度大于5的字符串。
val rdd = sc.parallelize(Seq("apple", "banana", "orange", "pear", "watermelon", "grape", "pineapple"))
import org.apache.spark.{Partitioner}
class MyPartitioner(numParts:Int) extends Partitioner{
  //覆盖分区数
  override def numPartitions: Int = numParts 
  //覆盖分区号获取函数
  override def getPartition(key: Any): Int = {
    val len:Int=key.toString.length
    len match{
      case s :Int => if (len <= 5) 0 else 1
    }
  }
}

val partitionedRdd=rdd.map(x=>(x,1)).partitionBy(new MyPartitioner(2))
val resultRDD = partitionedRdd.mapPartitions(iter => Iterator(iter.size))
// 打印计数结果
resultRDD.collect().foreach(println)
